import json
import logging

from confluent_kafka import Consumer

from src.constants import diseases, topics
from src.settings import KafkaSettings

logger = logging.getLogger(__name__)


class KafkaConsumer:
    def __init__(self, kafka_settings: KafkaSettings):
        self._kafka_settings = kafka_settings
        self.consumer = None
        self.consumer_config = {
            "bootstrap.servers": self._kafka_settings.boostrap_server,
            "group.id": self._kafka_settings.group_id,
            "sasl.mechanism": self._kafka_settings.sasl_mechanism,
            "security.protocol": self._kafka_settings.security_protocol,
            "sasl.username": self._kafka_settings.sasl_username,
            "sasl.password": self._kafka_settings.sasl_password,
            "enable.auto.commit": False,
            "auto.offset.reset": "earliest",
        }
        self.topics = topics
        self.diseases = diseases

    async def consume_messages(self):
        while True:
            msg = self.consumer.poll(1.0)

            if msg is None:
                continue

            topic = msg.topic()
            value = msg.value().decode("utf-8")

            logger.info("Got message")

            message_data = json.loads(value)
            disease_data = diseases[message_data["id"]] + ".from_message_data(message_data)"
            message = eval(disease_data)
            logger.info(f"Preprocessed message from {topic}: {message}")

    async def start(self):

        self.consumer = Consumer(self.consumer_config)

        self.consumer.subscribe(self.topics)

        await self.consume_messages()

        self.consumer.close()
